package com.example.demospringmybatisplus.service;

import com.alibaba.druid.util.StringUtils;
import com.example.demospringmybatisplus.handler.RefreshNodeRejectHandler;
import com.example.demospringmybatisplus.mapper.NodeMapper;
import com.example.demospringmybatisplus.vo.NodeVo;

import org.apache.commons.collections.CollectionUtils;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 异步刷新网元服务
 *
 * @author huahua
 * @DATE 2025/5/1
 **/
@Service
public class AsynFleshNodeService implements ApplicationRunner {
    private ThreadPoolExecutor threadPoolExecutor;

    @Autowired
    private RefreshNodeRejectHandler refreshNodeRejectHandler;

    @Autowired
    private RedissonClient redissionClient;

    @Autowired
    private NodeMapper nodeMapper;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        ThreadPoolExecutor singleThreadPoolExecutor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>(1000));
        singleThreadPoolExecutor.execute(this::initTask);
    }

    private void initTask() {
        if (!openFleshNode()) {
            return;
        }

        // 创建线程池，线程队列大小为10000，线程队列满时拒绝任务，拒绝策略为自定义的RefreshNodeRejectHandler
        threadPoolExecutor = new ThreadPoolExecutor(1, 4, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10000),
            Executors.defaultThreadFactory(), refreshNodeRejectHandler);

        while (executeRunningData()) {
            // 如果线程队列大于6000或者线程队列为空且线程数小于等于0，则线程休眠5s
            if (threadPoolExecutor.getQueue().size() >= 6000 || (threadPoolExecutor.getQueue().isEmpty()
                && threadPoolExecutor.getActiveCount() == 0)) {
                try {
                    // 线程休眠5s
                    long sleepTimes = 5000;
                    Thread.sleep(sleepTimes);
                } catch (InterruptedException e) {
                    System.out.println("线程休眠异常");
                }
            }
        }
    }

    private boolean openFleshNode() {
        // 从数据字典获取打开开关
        String isOpen = "Y";
        return StringUtils.equals(isOpen, "Y");
    }

    private boolean paushFleshNode() {
        // 从数据字典获取暂停开关
        String isPause = "N";
        return StringUtils.equals(isPause, "Y");
    }

    // 处理running数据，设置为true代表一直执行的方法
    protected boolean executeRunningData() {
        while (!paushFleshNode() && CollectionUtils.isNotEmpty(nodeMapper.getNodeList())) {
            // 这里锁名建议定义枚举类
            RLock lock = redissionClient.getLock("fleshNode");
            // nodeVoList从数据库获取，这里模拟从数据库获取数据
            List<NodeVo> nodeVoList = new ArrayList<>();
            try {
                // 设置等待时间10s，锁过期时间60s
                if (!lock.tryLock(10, 60, TimeUnit.SECONDS)) {
                    continue;
                }
                System.out.println("执行刷新网元逻辑，写入数据库");
            } catch (InterruptedException e) {
                System.out.println("获取锁失败");
            } finally {
                // 释放锁
                if (lock.isLocked() && lock.isHeldByCurrentThread()) {
                    lock.unlock();
                }
            }

            for (NodeVo nodeVo : nodeVoList) {
                threadPoolExecutor.execute(() -> { System.out.println("执行网元其它操作业务，写入数据库:" + nodeVo); });
            }
        }
        return openFleshNode();
    }
}
